#pragma once
#include <cassert>
#include <string>
#include <random>

#include <absl/time/clock.h>
#include <zmq.hpp>

#include "zsocket.hpp"
#include "mdp.h"

class mdclient
{
public:
	mdclient(std::string broker, bool verbose = false) :
		m_client(m_ctx, zmq::socket_type::req),
		m_retries(3),
		m_verbose(verbose),
		m_broker(broker),
		m_timeout(absl::Milliseconds(5000))
	{
		connect_to_broker();
	}
	~mdclient() 
	{
		m_client.close();
	}

public:
	void set_timeout(absl::Duration timeout)
	{
		m_timeout = timeout;
	}

	void set_retries(int retries)
	{
		m_retries = retries;
	}

public:
	zframe send(std::string service, zframe& request)
	{
		request.push_front(service);
		request.push_front(g_mdpc_client);
		if (m_verbose) {
			info("send request to {} service: ", service);
			request.dump();
		}

		zframe reply;
		int retries_left = m_retries;
		while (retries_left)
		{
			m_client.send(request.copy());
			zmq::pollitem_t items = { m_client, 0, ZMQ_POLLIN, 0 };
			zmq::poll(&items, 1, int(absl::ToInt64Milliseconds(m_timeout)));
			if (items.revents & ZMQ_POLLIN)
			{
				m_client.recv(reply);
				if (m_verbose) {
					info("received reply:");
					reply.dump();
				}

				assert(reply.size() >= 3);
				assert(reply.front() == g_mdpc_client);
				reply.pop_front();

				assert(reply.front() == service);
				reply.pop_front();
				return reply;
			}
			else if (--retries_left)
			{
				if (m_verbose) {
					info("no reply, reconnecting...");
				}
				connect_to_broker();
			}
			else
			{
				if (m_verbose) {
					info("permanent error, abandoning");
					break;
				}
			}
		}
		return reply;
	}

private:
	void connect_to_broker()
	{
		m_client = zsocket(m_ctx, zmq::socket_type::req);
		std::string id = m_client.set_random_id();
		m_client.connect(m_broker);
		m_client.set_linger(0);
		if (m_verbose) {
			info("{}: connecting to broker at {}", id, m_broker);
		}
	}

private:
	zmq::context_t m_ctx;
	zsocket m_client;

private:
	bool m_verbose;
	std::string m_broker;
	absl::Duration m_timeout;
	int m_retries;
};